package io.kiki.stack.netty.protocol;


import io.kiki.stack.netty.channelManager.Channel;
import io.kiki.stack.netty.command.CommandFactory;
import io.kiki.stack.netty.command.HeartbeatRequestCommand;
import io.kiki.stack.netty.command.ResponseCommand;
import io.kiki.stack.netty.command.ResponseType;
import io.kiki.stack.netty.config.ConfigManager;
import io.kiki.stack.netty.invoker.CallbackListener;
import io.kiki.stack.netty.invoker.Future;
import io.kiki.stack.netty.invoker.FutureImpl;
import io.kiki.stack.netty.util.InetSocketAddressUtil;
import io.kiki.stack.netty.util.TimerHolder;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.util.Timeout;
import io.netty.util.TimerTask;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.TimeUnit;

@Slf4j
@Setter
@Getter
public class RpcHeartbeatTrigger implements HeartbeatTrigger {
    /**
     * max trigger times
     */
    public static final Integer maxCount = ConfigManager.tcp_idle_maxtimes();

    private static final long heartbeatTimeoutMillis = 1000;

    private final CommandFactory commandFactory;

    public RpcHeartbeatTrigger(CommandFactory commandFactory) {
        this.commandFactory = commandFactory;
    }


    @Override
    public void heartbeatTriggered(final ChannelHandlerContext channelHandlerContext) throws Exception {
        Integer heartbeatTimes = channelHandlerContext.channel().attr(Channel.HEARTBEAT_COUNT).get();
        final Channel channel = channelHandlerContext.channel().attr(Channel.CHANNEL).get();
        if (heartbeatTimes >= maxCount) {
            try {
                channel.close();
                log.error("Heartbeat failed for {} times, close the Channel from client side: {} ", heartbeatTimes, InetSocketAddressUtil.parseRemoteAddress(channelHandlerContext.channel()));
            } catch (Exception e) {
                log.warn("Exception caught when closing Channel in SharableHandler.", e);
            }
        } else {
            boolean heartbeatSwitch = channelHandlerContext.channel().attr(Channel.HEARTBEAT_SWITCH).get();
            if (!heartbeatSwitch) {
                return;
            }
            final HeartbeatRequestCommand heartbeat = new HeartbeatRequestCommand();

            final Future future = new FutureImpl(heartbeat.getId(), new CallbackListener() {
                @Override
                public void onResponse(Future future) {
                    ResponseCommand response;
                    try {
                        response = (ResponseCommand) future.waitResponse(0);
                    } catch (InterruptedException e) {
                        log.error("Heartbeat ack process error! Id={}, from remoteAddr={}", heartbeat.getId(), InetSocketAddressUtil.parseRemoteAddress(channelHandlerContext.channel()), e);
                        return;
                    }
                    if (response != null && response.getResponseStatus() == ResponseType.success) {
                        if (log.isDebugEnabled()) {
                            log.debug("Heartbeat ack received! Id={}, from remoteAddr={}", response.getId(), InetSocketAddressUtil.parseRemoteAddress(channelHandlerContext.channel()));
                        }
                        channelHandlerContext.channel().attr(Channel.HEARTBEAT_COUNT).set(0);
                    } else {
                        if (response != null && response.getResponseStatus() == ResponseType.timeout) {
                            log.error("Heartbeat timeout! The address is {}", InetSocketAddressUtil.parseRemoteAddress(channelHandlerContext.channel()));
                        } else {
                            log.error("Heartbeat exception caught! Error code={}, The address is {}", response == null ? null : response.getResponseStatus(), InetSocketAddressUtil.parseRemoteAddress(channelHandlerContext.channel()));
                        }
                        Integer times = channelHandlerContext.channel().attr(Channel.HEARTBEAT_COUNT).get();
                        channelHandlerContext.channel().attr(Channel.HEARTBEAT_COUNT).set(times + 1);
                    }
                }

                @Override
                public String getRemoteAddress() {
                    return channelHandlerContext.channel().remoteAddress().toString();
                }
            }, null, this.commandFactory);
            final int heartbeatId = heartbeat.getId();
            channel.addFuture(future);
            if (log.isDebugEnabled()) {
                log.debug("Send heartbeat, successive count={}, Id={}, to remoteAddr={}", heartbeatTimes, heartbeatId, InetSocketAddressUtil.parseRemoteAddress(channelHandlerContext.channel()));
            }
            channelHandlerContext.writeAndFlush(heartbeat).addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture channelFuture) throws Exception {
                    if (channelFuture.isSuccess()) {
                        if (log.isDebugEnabled()) {
                            log.debug("Send heartbeat done! Id={}, to remoteAddr={}", heartbeatId, InetSocketAddressUtil.parseRemoteAddress(channelHandlerContext.channel()));
                        }
                    } else {
                        log.error("Send heartbeat failed! Id={}, to remoteAddr={}", heartbeatId, InetSocketAddressUtil.parseRemoteAddress(channelHandlerContext.channel()));
                    }
                }
            });
            TimerHolder.getTimer().newTimeout(new TimerTask() {
                @Override
                public void run(Timeout timeout) throws Exception {
                    Future future = channel.removeFuture(heartbeatId);
                    if (future != null) {
                        future.putResponse(commandFactory.createTimeoutResponse(channel.getRemoteAddress()));
                        future.tryAsyncExecuteInvokeCallbackAbnormally();
                    }
                }
            }, heartbeatTimeoutMillis, TimeUnit.MILLISECONDS);
        }

    }
}
